package org.example.mqtt;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.example.config.MqttConfig;
import org.example.util.constantUtil.ConstantUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Arrays;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;

@Component
@Slf4j
public class MqttService {
    private MqttConfig mqttConfig;
    public MqttClient client;

    // topic:imei_functionCode - 收到的回复结果
    private final ConcurrentHashMap<String, CompletableFuture<String>> responseMap = new ConcurrentHashMap<>();


    // 解决了循环依赖问题
    @Autowired
    public MqttService(MqttConfig mqttConfig) {
        this.mqttConfig = mqttConfig;
        initClient();
    }

    public void publishMessage(String topic, String message) throws MqttException {
        log.info("==========发送Mqtt消息 topic:{}, message:{}==========", topic, message);
        //  转换为hex编码
        byte[] data = new byte[message.length() / 2];
        for (int i = 0; i < data.length; i++) {
            int index = i * 2;
            int j = Integer.parseInt(message.substring(index, index + 2), 16);
            data[i] = (byte) j;
        }
        MqttMessage mqttMessage = new MqttMessage(data);
        client.publish(topic, mqttMessage);
    }
    // key为 imei_functionCode 的形式注册到responseMap
    public void registerCallback(String imei, int functionCode, CompletableFuture<String> future) {
        String key = imei + "_" + decimalToHex(functionCode);
        log.info("register key:{}", key);
        responseMap.put(key, future);
    }
    public void unregisterCallback(String imei, int functionCode) {
        String key = imei + "_" + decimalToHex(functionCode);
        log.info("unregister key:{}", key);
        responseMap.remove(key);
    }

    private void initClient(){
        MqttConnectOptions connectOptions = new MqttConnectOptions();
        //断开之后自动重联
        connectOptions.setAutomaticReconnect(true);
        //设置会话心跳时间 代理和客户端在不发送消息的情况下可以忍受的最长时间
        connectOptions.setKeepAliveInterval(mqttConfig.getKeepAlive());
        //不建立持久会话
        connectOptions.setCleanSession(true);
        //用户名
        connectOptions.setUserName(mqttConfig.getUsername());
        //密码
        connectOptions.setPassword(String.valueOf(mqttConfig.getPassword()).toCharArray());

        String clientId = MqttClient.generateClientId();

        try {
            client=new MqttClient(mqttConfig.getHost(), clientId, new MemoryPersistence());
            log.info("MQTT创建, host:{}, clientId:{}",mqttConfig.getHost(), clientId);
            // 消息回调
            client.setCallback(new MqttCallbackExtended() {
                @Override
                public void connectionLost(Throwable cause) {
                    log.error("==========MqttClient断开==========");
                    // todo 重连机制
                }

                @Override
                public void messageArrived(String topic, MqttMessage message) throws Exception {
                    // 加载到ConcurrentHashMap中对应的CompletableFuture对象中
                    log.info("收到MQTT消息, topic:{}, message:{}",topic, Arrays.toString(message.getPayload()));
                    handleResponse(topic, message);
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                    log.info("==========MQTT消息发送成功==========");
                }

                @Override
                public void connectComplete(boolean reconnect, String serverURI) {
                    log.info("==========MqttClient连接成功=============");
                    if (reconnect) {
                        log.info("Reconnect:重新订阅主题...");
                        subC1Topic();
                    }
                }
            });

            // MQTT连接
            client.connect(connectOptions);

            // 初始化订阅主题
            subC1Topic();

        } catch (MqttException e) {
            log.error("MQTT创建失败",e);
        }
    }



    // 订阅所有电池包对应的C1 Topic
    private void subC1Topic() {
        try {
            String topicSub = ConstantUtil.TOPIC_PREFIX + "+" + ConstantUtil.SUB_TOPIC_SUFFIX_CONTROL;
            client.subscribe(topicSub, mqttConfig.getQOS_1());
            log.info("MQTT订阅topic:{}", topicSub);
        } catch (MqttException e) {
            log.error("MQTT消息订阅异常", e);
        }
    }
    // 将MQTT消息加载到ConcurrentHashMap中对应的CompletableFuture对象中
    private void handleResponse(String topic, MqttMessage message) {
        // 获取消息对应的key
        byte[] dataBytes = message.getPayload();
        String key = topic.split("/")[1] + "_" + String.format("%02x", dataBytes[0]);
        // 返回到对应的结果
        log.info("handleResponse, key:{}", key);
        if (responseMap.containsKey(key)) {
            // 获取响应结果
            StringBuilder sb = new StringBuilder();
            for (byte dataByte : dataBytes) {
                sb.append(String.format("%02x", dataByte));
            }
            log.info("写入future, received data:{}", sb.toString());
            CompletableFuture<String> future = responseMap.get(key);
            future.complete(sb.toString());
        }
    }

    // 整数转换为1个字节的十六进制
    private static String decimalToHex(int decimal) {
        String hex = Integer.toHexString(decimal);
        return hex.length() == 1 ? "0" + hex : hex;
    }
}
